{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4083985e",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "预处理：使用 Spark 将数据转为 DataFrame\n",
    "首先需要将数据加载成 RDD，再将 RDD 转化为 DataFrame\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b828d58e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 初始化\n",
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import functions as F \n",
    "from pyspark.sql import SparkSession, Row  \n",
    "from pyspark.sql.types import StringType, StructField, StructType,IntegerType,FloatType  \n",
    "import os   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "224e4cf9",
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf()\n",
    "sc = SparkContext(conf=conf)\n",
    "sc.setLogLevel('WARN')\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "959eca86",
   "metadata": {},
   "outputs": [],
   "source": [
    "schemaString = \"Country,Year,Total,Coal,Oil,Gas,Cement,Flaring,Other,PerCapita\"\n",
    "fields = []\n",
    "for field in schemaString.split(\",\"):  \n",
    "    if field == 'Country':  \n",
    "        a = StructField(field, StringType(), True)  \n",
    "    elif field == 'Year':  \n",
    "        a = StructField(field,IntegerType(),True)  \n",
    "    else:  \n",
    "        a = StructField(field,FloatType(),True)  \n",
    "    fields.append(a)\n",
    "schema = StructType(fields)  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "8fe8cef4",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----+--------+--------+--------+--------+--------+-------+-----+---------+\n",
      "|    Country|Year|   Total|    Coal|     Oil|     Gas|  Cement|Flaring|Other|PerCapita|\n",
      "+-----------+----+--------+--------+--------+--------+--------+-------+-----+---------+\n",
      "|Afghanistan|1949|0.014656|0.014656|     0.0|     0.0|     0.0|    0.0|  0.0|      0.0|\n",
      "|Afghanistan|1950|0.084272|0.021068|0.063204|     0.0|     0.0|    0.0|  0.0| 0.011266|\n",
      "|Afghanistan|1951|  0.0916|0.025648|0.065952|     0.0|     0.0|    0.0|  0.0| 0.012098|\n",
      "|Afghanistan|1952|  0.0916|0.031708|0.059892|     0.0|     0.0|    0.0|  0.0| 0.011946|\n",
      "|Afghanistan|1953|0.106256|0.037949|0.068307|     0.0|     0.0|    0.0|  0.0| 0.013685|\n",
      "|Afghanistan|1954|0.106256|0.042502|0.063754|     0.0|     0.0|    0.0|  0.0| 0.013511|\n",
      "|Afghanistan|1955|0.153888|0.062288|  0.0916|     0.0|     0.0|    0.0|  0.0| 0.019304|\n",
      "|Afghanistan|1956|  0.1832|0.062288|0.120912|     0.0|     0.0|    0.0|  0.0| 0.022652|\n",
      "|Afghanistan|1957| 0.29312|0.076944|0.216176|     0.0|     0.0|    0.0|  0.0| 0.035702|\n",
      "|Afghanistan|1958| 0.32976|  0.0916| 0.23816|     0.0|     0.0|    0.0|  0.0| 0.039569|\n",
      "|Afghanistan|1959|0.384571| 0.10992| 0.25648|     0.0|0.018171|    0.0|  0.0| 0.045414|\n",
      "|Afghanistan|1960|0.413885|0.127115|0.268758|     0.0|0.018012|    0.0|  0.0| 0.048001|\n",
      "|Afghanistan|1961|0.490798|0.175872| 0.29312|     0.0|0.021806|    0.0|  0.0| 0.055835|\n",
      "|Afghanistan|1962|0.688594|0.296784|0.362736|     0.0|0.029074|    0.0|  0.0| 0.076775|\n",
      "|Afghanistan|1963|0.706736|0.263808|0.392048|     0.0| 0.05088|    0.0|  0.0| 0.077176|\n",
      "|Afghanistan|1964|0.838551|0.300448| 0.47632|     0.0|0.061783|    0.0|  0.0| 0.089632|\n",
      "|Afghanistan|1965|1.006917|0.381056|0.542272|     0.0|0.083589|    0.0|  0.0| 0.105269|\n",
      "|Afghanistan|1966|1.091159|0.428688|0.575248|     0.0|0.087223|    0.0|  0.0| 0.111535|\n",
      "|Afghanistan|1967|1.281865|0.399376|0.556928|0.260144|0.065417|    0.0|  0.0| 0.128058|\n",
      "|Afghanistan|1968|1.223391|0.332429|0.496817|0.347041|0.047105|    0.0|  0.0| 0.119381|\n",
      "+-----------+----+--------+--------+--------+--------+--------+-------+-----+---------+\n",
      "only showing top 20 rows\n",
      "\n",
      "root\n",
      " |-- Country: string (nullable = true)\n",
      " |-- Year: integer (nullable = true)\n",
      " |-- Total: float (nullable = true)\n",
      " |-- Coal: float (nullable = true)\n",
      " |-- Oil: float (nullable = true)\n",
      " |-- Gas: float (nullable = true)\n",
      " |-- Cement: float (nullable = true)\n",
      " |-- Flaring: float (nullable = true)\n",
      " |-- Other: float (nullable = true)\n",
      " |-- PerCapita: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rdd1 = sc.textFile('file:///C:/Users/Administrator/Desktop/demo/result.csv')  \n",
    "rdd2 = rdd1.map(lambda x:x.split(\"\\t\")). \\\n",
    "    map(lambda p: Row(p[0],int(p[1]),float(p[2]),float(p[3]),float(p[4]),float(p[5]),float(p[6]),float(p[7]),float(p[8]),float(p[9])))\n",
    "df = spark.createDataFrame(rdd2, schema)  \n",
    "df.createOrReplaceTempView(\"usInfo\")\n",
    "df.show() \n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "dc66b9e8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "|Country|Year|    Total|     Coal|    Oil|     Gas|   Cement| Flaring|    Other|PerCapita|\n",
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "|  China|2021|11472.369|7955.9854|1713.34|773.8661|852.96136|4.677478|171.53888| 8.045741|\n",
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "\n",
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "|Country|Year|    Total|     Coal|    Oil|     Gas|   Cement| Flaring|    Other|PerCapita|\n",
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "|  China|2021|11472.369|7955.9854|1713.34|773.8661|852.96136|4.677478|171.53888| 8.045741|\n",
      "+-------+----+---------+---------+-------+--------+---------+--------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#1. 查询2021年中国的排放数据\n",
    "df1 = spark.sql(\"\"\"\n",
    "    select * from usInfo where Country='China' and Year='2021'\n",
    "\"\"\")\n",
    "df1.show()\n",
    "\n",
    "df_01_01 = df.where(df['Country']=='China').where(df['Year']=='2021')\n",
    "df_01_01.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "98219df6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|      total_energy|\n",
      "+------------------+\n",
      "|1693867.3560400738|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|      total_energy|\n",
      "+------------------+\n",
      "|1693867.3560400738|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#2. 统计总CO2排放量\n",
    "df2 = df.selectExpr(\"sum(Total) as total_energy\")\n",
    "df2.show()\n",
    "\n",
    "df_01_02 = spark.sql(\"\"\"\n",
    "    select sum(Total) as total_energy from usInfo\n",
    "\"\"\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "924409f1",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|             Country|        AvgPerCapita|\n",
      "+--------------------+--------------------+\n",
      "|       Côte d'Ivoire| 0.41036192246247083|\n",
      "|                Chad|0.061354270160553004|\n",
      "|Micronesia (Feder...|  1.3467400650183359|\n",
      "|            Anguilla|     8.9916373193264|\n",
      "|            Kiribati|  0.4206226729467267|\n",
      "|              Guyana|  1.9942815370029874|\n",
      "|             Eritrea| 0.19605091082699158|\n",
      "|            Djibouti|   0.634266362629003|\n",
      "|                Fiji|  0.9536575136913193|\n",
      "|                Iraq|  2.1732026408327387|\n",
      "|     Leeward Islands| 0.21494871377944946|\n",
      "|             Germany|  5.7016681338457955|\n",
      "|             Comoros| 0.16571480869537308|\n",
      "|         Afghanistan|   0.149403301909668|\n",
      "|            Cambodia| 0.20180513462357558|\n",
      "|              Jordan|  2.1259180820650525|\n",
      "|              France|  3.7200810779756477|\n",
      "|              Greece|   2.866212268208041|\n",
      "|              Kosovo|   4.533541219575064|\n",
      "|British Virgin Is...|   4.533550151189169|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "+--------------------+--------------------+\n",
      "|             Country|      avg(PerCapita)|\n",
      "+--------------------+--------------------+\n",
      "|       Côte d'Ivoire| 0.41036192246247083|\n",
      "|                Chad|0.061354270160553004|\n",
      "|Micronesia (Feder...|  1.3467400650183359|\n",
      "|            Anguilla|     8.9916373193264|\n",
      "|            Kiribati|  0.4206226729467267|\n",
      "|              Guyana|  1.9942815370029874|\n",
      "|             Eritrea| 0.19605091082699158|\n",
      "|            Djibouti|   0.634266362629003|\n",
      "|                Fiji|  0.9536575136913193|\n",
      "|                Iraq|  2.1732026408327387|\n",
      "|     Leeward Islands| 0.21494871377944946|\n",
      "|             Germany|  5.7016681338457955|\n",
      "|             Comoros| 0.16571480869537308|\n",
      "|         Afghanistan|   0.149403301909668|\n",
      "|            Cambodia| 0.20180513462357558|\n",
      "|              Jordan|  2.1259180820650525|\n",
      "|              France|  3.7200810779756477|\n",
      "|              Greece|   2.866212268208041|\n",
      "|              Kosovo|   4.533541219575064|\n",
      "|British Virgin Is...|   4.533550151189169|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#3. 计算每个国家的平均人均能源使用量\n",
    "df3 = df.groupBy(\"Country\").agg({\"PerCapita\": \"avg\"}).withColumnRenamed(\"avg(PerCapita)\", \"AvgPerCapita\")  \n",
    "df3.show()\n",
    "df_01_03 = spark.sql(\"\"\"\n",
    "    select Country, avg(PerCapita) from usInfo group by Country\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "088c4970",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+---------+\n",
      "|Country|Year|    Total|\n",
      "+-------+----+---------+\n",
      "|  China|2021|11472.369|\n",
      "+-------+----+---------+\n",
      "only showing top 1 row\n",
      "\n",
      "排放总量最高的国家: China\n",
      "对应的年份: 2021\n",
      "+-------+----+---------+\n",
      "|Country|Year|    Total|\n",
      "+-------+----+---------+\n",
      "|  China|2021|11472.369|\n",
      "+-------+----+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#4. 获取排放总量最高的国家和对应的年份\n",
    "df4 = df.select(\"Country\", \"Year\", \"Total\").orderBy(df['Total'].desc())  \n",
    "df4.show(1) \n",
    "print(\"排放总量最高的国家:\", df4.first()[\"Country\"])  \n",
    "print(\"对应的年份:\", df4.first()[\"Year\"])  \n",
    "\n",
    "df_01_04 = spark.sql(\"\"\"\n",
    "    select Country, Year, Total from usInfo order by Total desc limit 1\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2ca8b611",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "|Country|Year|   Total|Coal|    Oil|     Gas|  Cement| Flaring|Other|PerCapita|\n",
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "|  Qatar|2021|95.66718| 0.0|8.67304|83.53126|1.571676|1.891205|  0.0|35.587357|\n",
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "only showing top 1 row\n",
      "\n",
      "2021年人均排放最高的国家: Qatar\n",
      "人均排放值: 35.58735656738281\n",
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "|Country|Year|   Total|Coal|    Oil|     Gas|  Cement| Flaring|Other|PerCapita|\n",
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "|  Qatar|2021|95.66718| 0.0|8.67304|83.53126|1.571676|1.891205|  0.0|35.587357|\n",
      "+-------+----+--------+----+-------+--------+--------+--------+-----+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#5. 2021年人均排放最高的国家\n",
    "df_2021 = df.filter(df.Year == 2021)  \n",
    "df5 = df_2021.orderBy(df_2021['PerCapita'].desc())  \n",
    "df5.show(1)  \n",
    "country = df5.first()[\"Country\"]  \n",
    "per_capita = df5.first()[\"PerCapita\"]  \n",
    "print(\"2021年人均排放最高的国家:\", country)  \n",
    "print(\"人均排放值:\", per_capita)\n",
    "\n",
    "df_01_05 = spark.sql(\"\"\"\n",
    "    select * from usInfo where Year='2021' order by PerCapita desc limit 1\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "362fd476",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+---------+---------+---------+---------+---------+--------+---------+---------+\n",
      "|Country|Year|    Total|     Coal|      Oil|      Gas|   Cement| Flaring|    Other|PerCapita|\n",
      "+-------+----+---------+---------+---------+---------+---------+--------+---------+---------+\n",
      "|  China|2000| 3644.464|2601.7368|648.68555| 59.79648| 244.3882|     0.0|89.857185| 2.883053|\n",
      "|  China|2001|3723.7307| 2637.655| 659.3002|67.157455| 270.8282|     0.0|    88.79|  2.92576|\n",
      "|  China|2002| 4112.459| 3018.634|  653.086|53.992702| 296.5903|     0.0|   90.156| 3.210536|\n",
      "|  China|2003| 4827.446| 3601.712| 727.0988|60.745457|345.18613|     0.0| 92.70359| 3.745477|\n",
      "|  China|2004| 5223.755|3835.2883|839.09265| 71.52128|379.99216|     0.0| 97.86024| 4.028136|\n",
      "|  China|2005| 5876.555|  4424.39|850.17255| 84.95717|411.64877|     0.0| 105.3869| 4.503496|\n",
      "|  China|2006|6488.8037| 4905.374| 898.5337|103.75349| 470.0866|     0.0| 111.0558| 4.941642|\n",
      "|  China|2007| 6978.612| 5289.904| 927.3511|129.51508|514.98083|     0.0| 116.8613| 5.280773|\n",
      "|  China|2008| 7496.832|5729.4736| 970.1173| 149.6744| 525.9245|     0.0| 121.6423| 5.636008|\n",
      "|  China|2009|7886.5327|   6028.5| 982.3477|164.40369|583.56024|     0.0|  127.721| 5.889315|\n",
      "|  China|2010| 8616.652|6568.7935|1092.5718| 199.5854| 639.5919|     0.0|   116.11| 6.391268|\n",
      "|  China|2011| 9528.556| 7309.786|1123.8588|245.98997| 708.5643|     0.0| 140.3565| 7.021286|\n",
      "|  China|2012| 9779.281|7464.8066|1176.7119| 277.5993| 714.7817|3.659815| 141.7225| 7.156126|\n",
      "|  China|2013| 9956.309|7493.4546| 1234.622|320.12747| 748.3231|3.564518|156.21687| 7.235162|\n",
      "|  China|2014| 9998.621| 7425.067|1255.6052|370.22156| 778.6271|3.877594|165.22316| 7.218233|\n",
      "|  China|2015| 9866.904| 7266.979| 1328.805|380.05222|721.99493|3.854226|165.21902| 7.079569|\n",
      "|  China|2016|9764.9795| 7071.404|1357.8375|421.72623|743.04407| 3.04058|167.92754| 6.965584|\n",
      "|  China|2017|10011.107| 7163.318|1430.5868|486.35938| 758.1852|2.924697|169.73322| 7.098687|\n",
      "|  China|2018|10353.877|   7316.4| 1500.474| 575.3689| 786.7446|3.350856|171.53888| 7.306542|\n",
      "|  China|2019|10740.996| 7543.157|1559.7281| 630.1714|826.87604|3.806248|177.25685| 7.554165|\n",
      "+-------+----+---------+---------+---------+---------+---------+--------+---------+---------+\n",
      "only showing top 20 rows\n",
      "\n",
      "+-------+----+---------+\n",
      "|Country|Year|    Total|\n",
      "+-------+----+---------+\n",
      "|  China|2021|11472.369|\n",
      "+-------+----+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#6. 中国历史排放总量（2000年后）\n",
    "df6 = df.filter((df['Country'] == \"China\") & (df['Year'] >= 2000))  \n",
    "df6.show()  \n",
    "\n",
    "df_01_06 = spark.sql(\"\"\"\n",
    "    select Country, Year, Total from usInfo where Year>'2020' and Country='China' order by PerCapita desc limit 1\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "17005b04",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+---------+\n",
      "|Year|Country|    Total|\n",
      "+----+-------+---------+\n",
      "|2021|  China|11472.369|\n",
      "|2020|  China|10956.213|\n",
      "|2019|  China|10740.996|\n",
      "|2018|  China|10353.877|\n",
      "|2017|  China|10011.107|\n",
      "|2016|  China|9764.9795|\n",
      "|2015|  China| 9866.904|\n",
      "|2014|  China| 9998.621|\n",
      "|2013|  China| 9956.309|\n",
      "|2012|  China| 9779.281|\n",
      "|2011|  China| 9528.556|\n",
      "|2010|  China| 8616.652|\n",
      "|2009|  China|7886.5327|\n",
      "|2008|  China| 7496.832|\n",
      "|2007|  China| 6978.612|\n",
      "|2006|  China|6488.8037|\n",
      "|2005|    USA|6137.6035|\n",
      "|2004|    USA| 6117.963|\n",
      "|2003|    USA| 6015.804|\n",
      "|2002|    USA|5952.6987|\n",
      "|2001|    USA|5911.9883|\n",
      "|2000|    USA|6016.3506|\n",
      "|1999|    USA|5810.3315|\n",
      "|1998|    USA|5737.1294|\n",
      "|1997|    USA|5691.8647|\n",
      "|1996|    USA|5616.4307|\n",
      "|1995|    USA|5427.7983|\n",
      "|1994|    USA|5365.5786|\n",
      "|1993|    USA| 5274.363|\n",
      "|1992|    USA|  5175.22|\n",
      "|1991|    USA|5064.9873|\n",
      "|1990|    USA| 5122.496|\n",
      "|1989|    USA|5132.1343|\n",
      "|1988|    USA|5050.4756|\n",
      "|1987|    USA| 4825.651|\n",
      "|1986|    USA|4663.3696|\n",
      "|1985|    USA|  4652.58|\n",
      "|1984|    USA|4662.1416|\n",
      "|1983|    USA| 4429.374|\n",
      "|1982|    USA| 4447.256|\n",
      "|1981|    USA| 4686.369|\n",
      "|1980|    USA|4808.5205|\n",
      "|1979|    USA|5008.5786|\n",
      "|1978|    USA| 4941.362|\n",
      "|1977|    USA|4889.6074|\n",
      "|1976|    USA| 4747.762|\n",
      "|1975|    USA| 4478.226|\n",
      "|1974|    USA| 4621.046|\n",
      "|1973|    USA| 4785.049|\n",
      "|1972|    USA| 4573.015|\n",
      "|1971|    USA|4365.4653|\n",
      "|1970|    USA| 4339.686|\n",
      "|1969|    USA|4035.1453|\n",
      "|1968|    USA|3840.9197|\n",
      "|1967|    USA| 3705.461|\n",
      "|1966|    USA|3571.4214|\n",
      "|1965|    USA|3399.5469|\n",
      "|1964|    USA|3264.3137|\n",
      "|1963|    USA|3126.4875|\n",
      "|1962|    USA| 2993.902|\n",
      "|1961|    USA|2886.8728|\n",
      "|1960|    USA|2897.3152|\n",
      "|1959|    USA| 2831.923|\n",
      "|1958|    USA|2747.1077|\n",
      "|1957|    USA|2835.7722|\n",
      "|1956|    USA|2859.9949|\n",
      "|1955|    USA| 2728.512|\n",
      "|1954|    USA|2489.4622|\n",
      "|1953|    USA|2612.9712|\n",
      "|1952|    USA|2551.2195|\n",
      "|1951|    USA|2618.7117|\n",
      "|1950|    USA|2541.4854|\n",
      "|1949|    USA|2164.8623|\n",
      "|1948|    USA|2582.5586|\n",
      "|1947|    USA|2485.8545|\n",
      "|1946|    USA|2257.4937|\n",
      "|1945|    USA|2359.5515|\n",
      "|1944|    USA|2444.9036|\n",
      "|1943|    USA| 2272.059|\n",
      "|1942|    USA| 2198.265|\n",
      "|1941|    USA| 2043.585|\n",
      "|1940|    USA|1874.9851|\n",
      "|1939|    USA|1670.9204|\n",
      "|1938|    USA|1515.7499|\n",
      "|1937|    USA|1792.6099|\n",
      "|1936|    USA|1713.4117|\n",
      "|1935|    USA|1492.3209|\n",
      "|1934|    USA|1436.2386|\n",
      "|1933|    USA|1348.8768|\n",
      "|1932|    USA|1257.8141|\n",
      "|1931|    USA|1484.9591|\n",
      "|1930|    USA|1744.9163|\n",
      "|1929|    USA|1963.5062|\n",
      "|1928|    USA| 1830.965|\n",
      "|1927|    USA|1862.8391|\n",
      "|1926|    USA|1906.0485|\n",
      "|1925|    USA|1758.1854|\n",
      "|1924|    USA|1698.3203|\n",
      "|1923|    USA|1896.7358|\n",
      "|1922|    USA|1435.7583|\n",
      "+----+-------+---------+\n",
      "only showing top 100 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#7. 查询每年排放总量最高的国家\n",
    "query = \"\"\" \n",
    "    SELECT Year, Country, Total \n",
    "    FROM ( \n",
    "        SELECT Year, Country, Total, ROW_NUMBER() OVER (PARTITION BY Year ORDER BY Total DESC) as rank \n",
    "        FROM usInfo \n",
    "    ) ranked \n",
    "    WHERE rank = 1 \n",
    "    ORDER BY Year desc\n",
    "\"\"\"  \n",
    "df7 = spark.sql(query)  \n",
    "df7.show(100) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "549e0633",
   "metadata": {},
   "outputs": [],
   "source": [
    "#8. 2021年排放前10的国家\n",
    "df_2021 = df.filter(df.Year == \"2021\")  \n",
    "df8 = df_2021.orderBy(df_2021['Total'].desc()).limit(10)  \n",
    "df8.show()\n",
    "\n",
    "df_01_08 = spark.sql(\"\"\"\n",
    "    select * from usInfo where Year='2021' order by Total desc limit 10\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "adfa13c8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
